package cn.easy.jacob.config;


import cn.easy.jacob.rocketmq.MessageListener;
import cn.easy.jacob.rocketmq.RocketMQManager;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.ArrayList;
import java.util.List;

@Configuration
public class RocketMQConfig {
    private final Logger LOGGER = LoggerFactory.getLogger(MQProducer.class);

    @Value("${apache.rocketmq.producer.producerGroup}")
    private String producerGroup;

    @Value("${apache.rocketmq.consumer.PushConsumer}")
    private String consumerGroup;

    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;

    @Bean(value = "mqProducer")
    public DefaultMQProducer defaultMQProducer() {
        try {
            LOGGER.info("消息生产者启动");
            DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
            producer.setNamesrvAddr(namesrvAddr);
//            producer.setVipChannelEnabled(false);
            producer.setRetryTimesWhenSendFailed(3);
            producer.start();
            return producer;
        } catch (MQClientException e) {
            LOGGER.error("MQ：启动生产者失败：{}-{}", e.getResponseCode(), e.getErrorMessage());
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    @Bean(value = "mqConsumer")
    public DefaultMQPushConsumer pushConsumer () throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
//        consumer.setVipChannelEnabled(false);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        return consumer;
    }

    @Bean("rocketMQManager")
    public RocketMQManager rocketMQManager1(MessageListener messageListener, DefaultMQPushConsumer mqConsumer) {
        List<MessageListener> messageListeners = new ArrayList<>();
        messageListeners.add(messageListener);
        return new RocketMQManager(mqConsumer, messageListeners, "zqh-message", "*");
    }
}
